/*************************************************************************
	> File Name:    dispatch_epoll.cpp
	> Author:       陈猛
	> Mail:         chenmeng@makeblock.com
	> Created Time: Thu 19 May 2022 01:30:45 AM PDT
	> Description:  
 ************************************************************************/
#include "common_log.h"
#include <netinet/in.h>
#include <sys/un.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include "dispatch_epoll.h"

using namespace dispatch;

#define SAFE_FREE(p)	do { if(p) { free(p);  p = NULL; }  }while(0)
#define SAFE_DELETE(p)	do { if(p) { delete p; p = NULL; }  }while(0)

#if _SYS_HAVE_EPOLL_HEADER_FILE_
EPoll::EPoll(const int32_t &nMaxFdNum)
    : m_epollfd(0)
    , m_nMaxFdNum(nMaxFdNum)
    , m_pEvents(NULL)
{
    m_pEvents = new struct epoll_event[nMaxFdNum];
    m_epollfd = epoll_create(nMaxFdNum);
}


EPoll::~EPoll()
{
    close(m_epollfd);

    SAFE_DELETE(m_pEvents);
}

#undef SAFE_FREE
#undef SAFE_DELETE

bool dispatch::EPoll::addFd(const int32_t &fd, const uint32_t &state, IEPoll* pData)
{
    struct epoll_event ev;
    ev.events = state;
    ev.data.ptr = pData;

    if(-1 == epoll_ctl(m_epollfd, EPOLL_CTL_ADD, fd, &ev))
    {
        if (EBADF==errno || EEXIST==errno || EINVAL==errno || ENOENT==errno || ENOMEM==errno || EPERM==errno)
        {
            GEN_Printf(LOG_ERROR, "epoll_ctl err: %s\n", strerror(errno));
            return false;
        }
    }

    return true;
}

bool EPoll::deleteFd(const int32_t &fd, IEPoll *)
{
    if(fd <= 0)
    {
        GEN_Printf(LOG_WARN, "occur fd <=0 while epoll delete fd, this will lead to memory leak!\n");
        return false;
    }

    if(-1 == epoll_ctl(m_epollfd, EPOLL_CTL_DEL, fd, NULL))
    {
        if(EBADF==errno || EEXIST==errno || EINVAL==errno || ENOENT==errno || ENOMEM==errno || EPERM==errno)
        {
            return true;
        }
    }

    return true;
}

bool EPoll::modifyFd(const int32_t &fd, const uint32_t &state, IEPoll *ptr)
{
    if(fd <= 0) return false;
#if 0
    struct epoll_event ev;
    ev.events   = state;
    ev.data.ptr = ptr;

    if(-1 == epoll_ctl(m_epollfd, EPOLL_CTL_ADD, fd, &ev))
    {
        if (EBADF==errno || EEXIST==errno || EINVAL==errno || ENOENT==errno || ENOMEM==errno || EPERM==errno)
        {
            GEN_Printf(LOG_ERROR, "epoll_ctl err: %s", strerror(errno));
            return false;
        }
    }
#else

    struct epoll_event ev;

    ev.events = state;
    ev.data.ptr = ptr;

    epoll_ctl(m_epollfd, EPOLL_CTL_DEL, fd, NULL);

    if(-1 == epoll_ctl(m_epollfd, EPOLL_CTL_ADD, fd, &ev))
    {
        if (EBADF==errno || EEXIST==errno || EINVAL==errno || ENOENT==errno || ENOMEM==errno || EPERM==errno)
        {
            GEN_Printf(LOG_ERROR, "epoll_ctl err: %s", strerror(errno));
            return false;
        }
    }
#endif
    return true;
}

int32_t EPoll::pollProc(int msec)
{
    int32_t ret = epoll_wait(m_epollfd, m_pEvents, m_nMaxFdNum, msec);

    // GEN_Printf(LOG_DEBUG, "epoll return %d event(s)", ret);
    for(int32_t i = 0; i < ret; ++i)
    {
        IEPoll* pData = (IEPoll*)(m_pEvents[i].data.ptr);

        if (m_pEvents[i].events&EPOLL_EVENT_READABLE && pData)
        {
            pData->read();
        }
        if (m_pEvents[i].events&EPOLL_EVENT_WRITEABLE && pData)
        {
            pData->write();
        }
        if (m_pEvents[i].events&EPOLL_EVENT_ERROR && pData)
        {
            //如果回调结果返回不等于成功，则忽略这个socket的其它事件
            pData->error();
        }
        if (m_pEvents[i].events&EPOLL_EVENT_HUP && pData)
        {
            pData->hup();
        }
        if (m_pEvents[i].events&EPOLL_EVENT_URGENT && pData)
        {
            pData->readUrgent();
        }
    }

    return 0;
}
#else
namespace dispatch {
typedef IEPoll*  pIEPoll ;
class PollPrivate
{
public:
    PollPrivate(int nMax)
        : mArrAlloc(nMax)
        , mArrSize(0)
        , mArrProc(new pIEPoll[nMax])
        , mArray(new pollfd[nMax])
    {}


    void append(int32_t fd, uint32_t state, IEPoll* pData)
    {
        if (fd < 0) return;
        // 需要防止重复添加
        for (unsigned i = 0; i < mArrSize; ++i)
        {
            if (mArrProc[i] == pData)
            {
                mArray[mArrSize].fd      = fd;
                mArray[mArrSize].events  = state;
                mArray[mArrSize].revents = 0;
                return;
            }
        }

        SF_ASSERT(mArrSize < mArrAlloc);

        mArray[mArrSize].fd      = fd;
        mArray[mArrSize].events  = state;
        mArray[mArrSize].revents = 0;
        mArrProc[mArrSize]       = pData;
        ++mArrSize;
    }
    void clear(int32_t fd, IEPoll *)
    {
        if (fd < 0) return;
        for (unsigned int i = 0; i < mArrSize; ++i)
        {
            if (fd == mArray[i].fd)
            {
                for(unsigned int j = i; j < mArrSize - 1; ++j)
                {
                    mArray[j]   = mArray[j + 1];
                    mArrProc[j] = mArrProc[j + 1];
                }
                --mArrSize;
            }
        }
    }

    uint32_t     mArrAlloc;
    uint32_t     mArrSize;

    pIEPoll     *mArrProc;
    pollfd      *mArray;
};

}
EPoll::EPoll(const int32_t& nMaxFdNum)
    : d(new PollPrivate(nMaxFdNum))
{
}
EPoll::~EPoll()
{
    delete d;
}
bool dispatch::EPoll::addFd(const int32_t &fd, const uint32_t &state, IEPoll* pData)
{
    d->append(fd, state, pData);
    return true;
}

bool EPoll::deleteFd(const int32_t &fd, IEPoll *pData)
{
    d->clear(fd, pData);
    return true;
}

bool EPoll::modifyFd(const int32_t &fd, const uint32_t &state, IEPoll *ptr)
{
    d->append(fd, state, ptr);
    return true;
}

int32_t EPoll::pollProc(int msec)
{
    pollfd *pfds = d->mArray;
    int ret  = poll(pfds, d->mArrSize, msec);
    for (uint32_t i = 0; ret && i < d->mArrSize; i++)
    {
        GEN_Printf(LOG_DEBUG, "revents=%#x, [%d]=%d", pfds[i].revents, i, ret);
        if (pfds[i].revents == 0)
        {
            continue;
        }

        IEPoll* pData = d->mArrProc[i];

        if (pfds[i].revents&EPOLL_EVENT_READABLE && pData)
        {
            pData->read();
        }
        if (pfds[i].revents&EPOLL_EVENT_WRITEABLE && pData)
        {
            pData->write();
        }
        if (pfds[i].revents&EPOLL_EVENT_ERROR && pData)
        {
            //如果回调结果返回不等于成功，则忽略这个socket的其它事件
            pData->error();
        }
        if (pfds[i].revents&EPOLL_EVENT_HUP && pData)
        {
            pData->hup();
        }
        if (pfds[i].revents&EPOLL_EVENT_URGENT && pData)
        {
            pData->readUrgent();
        }
        --ret;
    }

    return 0;
}
#endif // _SYS_HAVE_EPOLL_HEADER_FILE_


bool EPoll::selectSock(int msec, IEPoll *ptr)
{
    bool retval = false;
    int32_t ret = epoll_wait(m_epollfd, m_pEvents, m_nMaxFdNum, msec);
    for(int32_t i = 0; i < ret; ++i)
    {
        IEPoll* pData = (IEPoll*)(m_pEvents[i].data.ptr);
        if (m_pEvents[i].events&EPOLL_EVENT_READABLE)
        {
            if (!ptr || (ptr && ptr == pData))
            {
                retval = true;
                break;
            }
        }
    }
    return retval;
}

bool EPoll::selectWRSock(int msec, IEPoll *ptr)
{
    bool retval = false;
    int32_t ret = epoll_wait(m_epollfd, m_pEvents, m_nMaxFdNum, msec);
    for(int32_t i = 0; i < ret; ++i)
    {
        IEPoll* pData = (IEPoll*)(m_pEvents[i].data.ptr);
        if (m_pEvents[i].events&EPOLL_EVENT_WRITEABLE && pData)
        {
            if (!ptr || (ptr && ptr == pData))
            {
                retval = true;
                break;
            }
        }
    }
    return retval;
}

LocalEventSock::LocalEventSock()
    : m_monitor(-1)
    , m_notify(-1)
{
}

LocalEventSock::~LocalEventSock()
{
    ::close(m_monitor);
    ::close(m_notify);
}

bool LocalEventSock::addInEPoll(EPoll *ep)
{
    struct sockaddr_un addr;
    static int index = 0;

    memset(&addr, 0, sizeof(struct sockaddr_un));

    m_monitor = socket(PF_UNIX, SOCK_DGRAM, 0);
    SF_ASSERT(m_monitor > 0);
    addr.sun_family = AF_UNIX;
    do {
        snprintf(addr.sun_path, sizeof(addr.sun_path) -1, "/tmp/.LocalEventSock.%d.%d", getpid(), index++);
        // GEN_Printf(LOG_DEBUG, "----------------------");
    } while (0 == access(addr.sun_path, F_OK)); // =0存在就继续，找下一个；!=0不存在就退出

    //GEN_Printf(LOG_DEBUG, "%s", addr.sun_path);
    SF_ASSERT(0 == ::bind(m_monitor, (struct sockaddr *) &addr, sizeof(addr)));

    m_notify = socket(PF_UNIX, SOCK_DGRAM, 0);
    SF_ASSERT(m_notify > 0);
    SF_ASSERT(0 == ::connect(m_notify, (struct sockaddr *) &addr, sizeof(addr)));

    ep->addFd(m_monitor, EPOLL_EVENT_READABLE|EPOLL_EVENT_ERROR|EPOLL_EVENT_HUP, this);

    m_epoll = ep;

    ::remove(addr.sun_path);

    return true;
}

namespace {
struct LocalMsg_{
    int   sig;
    void *ptr;
};
}
void LocalEventSock::wakeUp()
{
    LocalMsg_ msg {0, 0};
    ::write(m_notify, &msg, sizeof(LocalMsg_));
}

void LocalEventSock::postEvent(int sig, void *ptr)
{
    LocalMsg_ msg {sig, ptr};
    ::write(m_notify, &msg, sizeof(LocalMsg_));
}

void LocalEventSock::read()
{
    char buf[sizeof(LocalMsg_) * 4];
    LocalMsg_ *event;
    int len, nread;
    len = ::read(m_monitor, buf, sizeof(buf));
    if (len > 0)
    {
        nread = 0;

        while (len > 0)
        {
            event = (LocalMsg_ *)&buf[nread];
            if (event->sig < MSG_INNER_MAX)
            {
                // GEN_Printf(LOG_DEBUG, "inner msg: %d, %p", event->sig, event->ptr);
            } else
            {
                Event evt(event->sig, event->ptr);
                runCb(evt);
            }

            nread = nread + sizeof(LocalMsg_);
            len = len - sizeof(LocalMsg_);
        }
    }
}

void LocalEventSock::error()
{
    GEN_Printf(LOG_ERROR, "monitor error.");
    m_epoll->deleteFd(m_monitor, this);
}

void LocalEventSock::hup()
{
    GEN_Printf(LOG_ERROR, "monitor hup.");
    m_epoll->deleteFd(m_monitor, this);
}

void LocalEventSock::destory()
{
    if (m_monitor > 0) { ::close(m_monitor); m_monitor = -1; }
    if (m_notify > 0) { ::close(m_notify); m_notify = -1; }
}

void IEPoll::error()
{
    GEN_Printf(LOG_ERROR, "INotify err.");
    SF_ASSERT(m_epoll);
}

void IEPoll::hup()
{
    GEN_Printf(LOG_ERROR, "INotify hup.");
    SF_ASSERT(m_epoll);
}
